package chapter04

/**
 * author: yuhui
 * descriptions: cache 和 persist 案例
 * date: 2025 - 02 - 12 11:04 上午
 */
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.serializer.KryoSerializer

object CacheExample {
  def main(args: Array[String]): Unit = {
    // 创建Spark配置和上下文
    val conf = new SparkConf()
      .setAppName("CacheExample")
      .set("spark.serializer", classOf[KryoSerializer].getName)
    val sc = new SparkContext(conf)

    // 读取数据
    val data = sc.textFile("hdfs:///path/to/data")

    // 对数据进行缓存，使用默认的MEMORY_ONLY存储级别
    data.cache()

    // 第一次Action操作，触发数据读取和缓存
    val result1 = data.filter(line => line.contains("error")).count()
    println(s"Number of error lines: $result1")

    // 第二次Action操作，直接从缓存中读取数据，无需重新计算
    val result2 = data.filter(line => line.contains("warning")).count()
    println(s"Number of warning lines: $result2")

    // 使用persist指定存储级别
    data.persist(org.apache.spark.storage.StorageLevel.MEMORY_AND_DISK_SER)

    // 注意：在Scala中，如果后续没有更多的Action操作，
    // Spark作业可能会因为没有触发执行计划而提前结束。
    // 为了演示目的，这里我们不再添加额外的Action操作，
    // 但在实际应用中，您应该根据需要添加更多的数据处理逻辑。

    // 停止SparkContext
    sc.stop()
  }
}